/*
 * Copyright 2009 Armando Blancas
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package taskgraph.tasks.data;

import java.io.EOFException;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.util.Collections;
import java.util.Set;

import taskgraph.ports.InputPort;
import taskgraph.ports.IntOutputPort;
import taskgraph.ports.OutputPort;
import taskgraph.tasks.PrimitiveTask;


/**
 * Reports distinct, unique or repeated objects, with or without counts.
 * 
 * <p>The default command is to transmit distinct objects, with a single 
 * entry when there are duplicate. If {@code COUNT} is requested and an
 * integer output port is provided, this task will write a count for each
 * object reference it writes to the {@code output} port.
 * 
 * @author Armando Blancas
 * @see taskgraph.ports.InputPort
 * @see taskgraph.ports.OutputPort
 * @see taskgraph.ports.IntOutputPort
 * @param <E> The type of elements filtered by this task.
 */
public class Unique<E> extends PrimitiveTask {
	
	public enum UNIQ_F { UNIQUE, REPEATED, COUNT };

	private InputPort<E> input;
	private OutputPort<E> output;
	private IntOutputPort countsPort;
	private Set<UNIQ_F> flags = Collections.emptySet();
	
	/**
	 * This constructor allows creating instances as beans.
	 */
    public Unique() {
    	// do nothing
    }

	/**
     * Creates an instance for selecting distinct objects.
     * 
     * @param input The input port.
     * @param output The output port.
     */
    public Unique(final InputPort<E> input, 
    		      final OutputPort<E> output) {
    	setInput(input);
    	setOutput(output);
    }

	/**
     * Creates a fully configured instance.
     * 
     * <p>If {@code flags} contains {@code COUNT} this task will write 
     * a count to {@code countsPort} for every object reference written 
     * to port {@code output}. Port {@code countsPort} may be null, in
     * which case this task will not attempt to use it.
     * 
     * @param input The input port.
     * @param output The output port.
     * @param countsPort The output port for sending counts.
     * @param flags The set of choices.
     */
    public Unique(final InputPort<E> input, 
    		      final OutputPort<E> output,
    		      final IntOutputPort countsPort,
    		      final Set<UNIQ_F> flags) {
    	setInput(input);
    	setOutput(output);
    	setCountsPort(countsPort);
    	setFlags(flags);
    }

    /**
	 * Gets the port used for sending counts.
	 * 
	 * @return The countsPort object
	 */
	public IntOutputPort getCountsPort() {
		return countsPort;
	}
	
    /**
	 * Gets the set of command options.
	 * 
	 * @return The set of flags.
	 */
	public Set<UNIQ_F> getFlags() {
		return flags;
	}
	
	/**
     * Gets the input port.
     * 
	 * @return The input port.
	 */
	public InputPort<E> getInput() {
		return input;
	}
	
	/**
     * Gets the output port.
     * 
	 * @return The output port.
	 */
	public OutputPort<E> getOutput() {
		return output;
	}
	
	/**
     * Reads lines from the input port and writes repeated or non-repeated 
     * lines, with or without counts, as requested.
	 *<pre>
	 *while not EOF
	 *    read an object from input port
	 *    report the object as requested
	 *close input port
	 *close output port
	 *</pre>
	 */
	@Override
    public void run() {
    	try {
    		boolean unique    = flags.contains(UNIQ_F.UNIQUE);
    		boolean repeated  = flags.contains(UNIQ_F.REPEATED);
    		boolean countAll  = !(unique ||repeated);
    		boolean keepCount = (flags.contains(UNIQ_F.COUNT)) && 
    		                    (countsPort != null);

    		E last = input.read();    		
    		int count = 1;    		
    		
    		while (!Thread.currentThread().isInterrupted()) {
    			
        		E element;
        		
        		/*
        		 * The algorithm goes one last pass after EOF, thus we 
        		 * must catch it here and indicate EOF with a null.
        		 */
        		try {
        			element = input.read();
        		} catch (EOFException e) {
                	element = null;
                }
    			
    			if (element != null && element.equals(last)) {
    				count++;
    			} else {
    				boolean printIt = countAll;
    				if (unique) {
    					printIt = (count == 1);
    				}
    				else if (repeated) {
    					printIt = (count > 1);
    				}
    				if (printIt) {
    					if (keepCount) {
    						countsPort.write(count);
    					}
    					output.write(last);
    				}
    				
    				if (element == null) {
    					break;
    				}
    				
    				count = 1;
       				last = element;
    			}
            } 
        } catch (InterruptedIOException e) {
        	// interruption request
        } catch (EOFException e) {
            log.error("No input for Unique#run()", e);
        } catch (IOException e) {
        	// pipe synch error
            log.error("Unique#run()", e);
        } finally {
			input.close();
			output.close();
        }
    }

	/**
	 * Sets the port used for sending counts.
	 * 
	 * <p>The argument may be null in order to clear a previously set value.
	 * 
	 * @param countsPort The countsPort to set
	 */
	public void setCountsPort(final IntOutputPort countsPort) {
		this.countsPort = countsPort;
	}

	/**
	 * Sets the commands options for counting.
	 * 
	 * @param flags The set of flags.
	 */
	public void setFlags(Set<UNIQ_F> flags) {
    	if (flags == null)
    		throw new IllegalArgumentException("flags == null");
    	if (flags.contains(UNIQ_F.UNIQUE) && flags.contains(UNIQ_F.REPEATED))
    		throw new IllegalArgumentException("cannot do unique and repeated");    	
		this.flags = flags;
	}

	/**
	 * Sets the input port.
	 * 
	 * @param input The input port.
	 */
    public void setInput(final InputPort<E> input) {
    	if (input == null)
    		throw new IllegalArgumentException("input == null");
 		this.input = input;
	}

	/**
	 * Sets the output port.
	 * 
	 * @param output The output port.
	 */
    public void setOutput(final OutputPort<E> output) {
    	if (output == null)
    		throw new IllegalArgumentException("output == null");
 		this.output = output;
	}

}
